/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.pubsub;

import org.redisson.PubSubEntry;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.RedisPubSubListener;
import org.redisson.client.codec.LongCodec;
import org.redisson.misc.AsyncSemaphore;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 声明一个抽象的发布/订阅类，该类用于管理发布和订阅事件,泛型E，它必须是PubSubEntry的子类型。
 * 这种设计模式允许发布者向订阅者发送消息，而无需两者之间有直接的耦合。
 * @param <E> 泛型参数，代表事件类型，该事件必须是PubSubEntry的子类型。
 */
abstract class PublishSubscribe<E extends PubSubEntry<E>> {


    /*
     * 基本概念
     * 1、LockPubSub            用于监听分布式锁的状态变化，当锁被释放时通知等待的线程去竞争获取锁
     * 2、SemaphorePubSub       与LockPubSub类似，但用于信号量，允许一定数量的并发访问，当资源可用时，通知等待的线程可以执行操作
     * 3、CountDownLatchPubSub  用于实现计数器的功能，当计数器达到零时，所有等待的线程将被唤醒继续执行
     *
     * 功能用途
     * 1、LockPubSub            主要用于实现对共享资源的互斥访问，确保任何时候只有一个客户端可以持有锁
     * 2、SemaphorePubSub       用于控制同时访问某一或多个资源的线程数量，适用于限流和速率控制的场景
     * 3、CountDownLatchPubSub  用于实现计数器，当计数器达到零时，所有等待的线程将被唤醒继续执行
     *
     * 使用场景
     * 1、LockPubSub            适用于需要严格控制资源访问顺序的高竞争场景，如防止库存超卖等
     * 2、SemaphorePubSub       适用于限制访问某个资源或服务的流量，避免过载
     * 3、CountDownLatchPubSub  适用于多线程协作，需要等待其他线程完成后再进行下一步的场景
     *
     *
     * 内部机制
     * 1、LockPubSub            通过订阅解锁事件，一旦锁被释放，等待的线程会收到通知并尝试获取锁。
     * 2、SemaphorePubSub       类似于LockPubSub，但关注的是信号量的释放，允许多个线程同时获取访问权
     * 3、CountDownLatchPubSub  通过计数器的减少来通知线程，当计数器为零时，表示所有需要等待的操作已完成
     *
     *
     * 性能考量
     * 1、LockPubSub             性能较高，适用于高竞争场景，但需要关注锁的释放和获取操作。
     * 2、SemaphorePubSub        性能较高，适用于高并发场景，但需要关注信号量的释放和获取操作。
     * 3、CountDownLatchPubSub   性能较高，适用于多线程协作场景，但需要关注计数器的减少和等待操作。
     *
     */



    /**
     * 存储订阅者信息
     */
    private final ConcurrentMap<String, E> entries = new ConcurrentHashMap<>();

    /**
     * 发布订阅服务
     */
    private final PublishSubscribeService service;

    PublishSubscribe(PublishSubscribeService service) {
        super();
        this.service = service;
    }


    /**
     * 添加频道订阅
     */
    public CompletableFuture<E> subscribe(String entryName, String channelName) {
        AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
        CompletableFuture<E> newPromise = new CompletableFuture<>();
        // 异步获取信号量结束后,thenAccept --> 尝试获取许可，若成功则进行订阅逻辑
        semaphore.acquire().thenAccept(c -> {

            if (newPromise.isDone()) {
                semaphore.release();
                return;
            }

            // 尝试从entries中获取已存在的条目
            E entry = entries.get(entryName);
            if (entry != null) {
                // 如果找到，获取该条目并释放semaphore
                entry.acquire();
                semaphore.release();
                // 完成newPromise，使用该条目的结果
                entry.getPromise().whenComplete((r, e) -> {
                    if (e != null) {
                        newPromise.completeExceptionally(e);
                        return;
                    }
                    newPromise.complete(r);
                });
                return;
            }

            // 如果未找到条目，创建一个新的条目并尝试获取锁
            E value = createEntry(newPromise);
            value.acquire();

            // 尝试将新条目放入entries，如果成功则处理，失败说明有其他线程已添加
            E oldValue = entries.putIfAbsent(entryName, value);
            if (oldValue != null) {
                // 如果存在旧条目，则使用旧条目进行处理
                oldValue.acquire();
                semaphore.release();
                oldValue.getPromise().whenComplete((r, e) -> {
                    if (e != null) {
                        newPromise.completeExceptionally(e);
                        return;
                    }
                    newPromise.complete(r);
                });
                return;
            }

            // 创建监听器并尝试订阅频道
            RedisPubSubListener<Object> listener = createListener(channelName, value);
            CompletableFuture<PubSubConnectionEntry> s = service.subscribeNoTimeout(LongCodec.INSTANCE, channelName, semaphore, listener);
            // 在订阅完成或出错时处理
            newPromise.whenComplete((r, e) -> {
                if (e != null) {
                    s.completeExceptionally(e);
                }
            });
            // 在订阅成功或出错时进行后续处理
            s.whenComplete((r, e) -> {
                if (e != null) {
                    // 如果出错，则从entries中移除条目，并完成value的promise异常
                    entries.remove(entryName);
                    value.getPromise().completeExceptionally(e);
                    return;
                }
                // 订阅成功，完成value的promise
                value.getPromise().complete(value);
            });

        });

        return newPromise;
    }


    /**
     * 取消频道订阅
     */
    public void unsubscribe(E entry, String entryName, String channelName) {
        ChannelName cn = new ChannelName(channelName);
        AsyncSemaphore semaphore = service.getSemaphore(cn);
        semaphore.acquire().thenAccept(c -> {
            if (entry.release() == 0) {
                entries.remove(entryName);
                service.unsubscribeLocked(cn)
                        .whenComplete((r, e) -> {
                            semaphore.release();
                        });
            } else {
                semaphore.release();
            }
        });
    }


    /**
     * 创建监听
     */
    private RedisPubSubListener<Object> createListener(String channelName, E value) {
        RedisPubSubListener<Object> listener = new BaseRedisPubSubListener() {

            @Override
            public void onMessage(CharSequence channel, Object message) {
                if (!channelName.equals(channel.toString())) {
                    return;
                }

                PublishSubscribe.this.onMessage(value, (Long) message);
            }
        };
        return listener;
    }


    public void timeout(CompletableFuture<?> promise) {
        service.timeout(promise);
    }

    public void timeout(CompletableFuture<?> promise, long timeout) {
        service.timeout(promise, timeout);
    }


    protected abstract E createEntry(CompletableFuture<E> newPromise);

    protected abstract void onMessage(E value, Long message);



}
